package com.atguigu.day7.procwindow

import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.types.Row
import org.apache.kafka.clients.producer.KafkaProducer

object SQLProcTime {




    def main(args: Array[String]): Unit = {

      val env = StreamExecutionEnvironment.getExecutionEnvironment
      env.setParallelism(1)

      val settings = EnvironmentSettings
        .newInstance()
        .useBlinkPlanner()
        .inStreamingMode()
        .build()


      val tEnv = StreamTableEnvironment.create(env, settings)
      val stream: DataStream[SensorReading] = env.addSource(new SensorSource)


      //'pt.proctime指定处理时间是'pt字段
      tEnv.createTemporaryView("sensor",stream,'id, 'timeStamP as 'ts, 'temperature, 'pt.proctime)

      tEnv.sqlQuery("SELECT id, COUNT(id) FROM sensor GROUP BY id, TUMBLE(pt,INTERVAL '10' SECOND)")
        .toAppendStream[Row]
        .print()




      env.execute()


    }
  }
